{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8bc47910-45c3-433e-a865-a6ef4a7af366",
   "metadata": {},
   "source": [
    "## A02_Data Formatting Process\n",
    "\n",
    "This code will convert the raw data into a formatted DataFrame and store it into a formatted Parquet file."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf3f9e86-9aaf-4886-8b53-19895cee8529",
   "metadata": {},
   "source": [
    "### 1. init spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "f4d85c4c-9c2c-4975-8c51-4837169df20e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "627b1415-b963-4ecf-b6f9-be9844eb56c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.appName(\"unemployment data\").getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e2954b3-47c7-40f4-bc98-4e61c1e83229",
   "metadata": {},
   "source": [
    "### 2. read the json data files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "d08a99aa-a77c-4bd5-81d8-1c4acdefc793",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_dir = \"./Landing Zone/unemployment\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "d16b4b9a-8386-41e2-9a91-20449eb1e52c",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.json(data_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "eabf3844-4a3f-495e-af14-475ed518b89b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+--------------------+-------+\n",
      "|               error|                help|              result|success|\n",
      "+--------------------+--------------------+--------------------+-------+\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|                NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "|{Not Found Error,...|https://opendata-...|                NULL|  false|\n",
      "+--------------------+--------------------+--------------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "437ba981-5e9f-4acb-951f-a4c9bee275bd",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- error: struct (nullable = true)\n",
      " |    |-- __type: string (nullable = true)\n",
      " |    |-- message: string (nullable = true)\n",
      " |-- help: string (nullable = true)\n",
      " |-- result: struct (nullable = true)\n",
      " |    |-- _links: struct (nullable = true)\n",
      " |    |    |-- next: string (nullable = true)\n",
      " |    |    |-- start: string (nullable = true)\n",
      " |    |-- fields: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- id: string (nullable = true)\n",
      " |    |    |    |-- type: string (nullable = true)\n",
      " |    |-- records: array (nullable = true)\n",
      " |    |    |-- element: struct (containsNull = true)\n",
      " |    |    |    |-- Any: string (nullable = true)\n",
      " |    |    |    |-- Codi_Barri: string (nullable = true)\n",
      " |    |    |    |-- Codi_Districte: string (nullable = true)\n",
      " |    |    |    |-- Demanda_ocupacio: string (nullable = true)\n",
      " |    |    |    |-- Demanda_ocupació: string (nullable = true)\n",
      " |    |    |    |-- Mes: string (nullable = true)\n",
      " |    |    |    |-- Nom_Barri: string (nullable = true)\n",
      " |    |    |    |-- Nom_Districte: string (nullable = true)\n",
      " |    |    |    |-- Nombre: string (nullable = true)\n",
      " |    |    |    |-- Sexe: string (nullable = true)\n",
      " |    |    |    |-- _id: long (nullable = true)\n",
      " |    |-- resource_id: string (nullable = true)\n",
      " |    |-- total: long (nullable = true)\n",
      " |-- success: boolean (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34bc8bb6-b071-42e7-999b-42f9d2fb74ad",
   "metadata": {},
   "source": [
    "### 3. filter the data which success=true"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "3f960101-b3d3-4562-bba3-d9da83b8fc63",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df[df[\"success\"]]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "d0e40e31-eafe-4c57-8d7c-88497c133390",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+--------------------+--------------------+-------+\n",
      "|error|                help|              result|success|\n",
      "+-----+--------------------+--------------------+-------+\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "| NULL|https://opendata-...|{{/api/action/dat...|   true|\n",
      "+-----+--------------------+--------------------+-------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7f3a2a51-082d-4bbc-bab2-0656f6cc3f5d",
   "metadata": {},
   "source": [
    "### 4. parse the records values to dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "843f4733-34ea-47c3-83eb-6c83977224a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import explode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "e3ee4b92-2e46-41af-9beb-a98f31db282b",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_new = df.select(explode(\"result.records\").alias(\"record\")).select(\"record.*\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "9bb8fc89-4572-4f6e-8d0f-0ba27ff85930",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+----------+--------------+----------------+----------------+---+--------------------+-------------+------+-----+---+\n",
      "| Any|Codi_Barri|Codi_Districte|Demanda_ocupacio|Demanda_ocupació|Mes|           Nom_Barri|Nom_Districte|Nombre| Sexe|_id|\n",
      "+----+----------+--------------+----------------+----------------+---+--------------------+-------------+------+-----+---+\n",
      "|2016|         1|             1|            NULL|  Atur registrat|  1|            el Raval| Ciutat Vella|  2431|Homes|  1|\n",
      "|2016|         2|             1|            NULL|  Atur registrat|  1|      el Barri Gòtic| Ciutat Vella|   588|Homes|  2|\n",
      "|2016|         3|             1|            NULL|  Atur registrat|  1|      la Barceloneta| Ciutat Vella|   637|Homes|  3|\n",
      "|2016|         4|             1|            NULL|  Atur registrat|  1|Sant Pere, Santa ...| Ciutat Vella|   878|Homes|  4|\n",
      "|2016|         5|             2|            NULL|  Atur registrat|  1|       el Fort Pienc|     Eixample|   693|Homes|  5|\n",
      "|2016|         6|             2|            NULL|  Atur registrat|  1|  la Sagrada Família|     Eixample|  1154|Homes|  6|\n",
      "|2016|         7|             2|            NULL|  Atur registrat|  1|la Dreta de l'Eix...|     Eixample|   772|Homes|  7|\n",
      "|2016|         8|             2|            NULL|  Atur registrat|  1|l'Antiga Esquerra...|     Eixample|   855|Homes|  8|\n",
      "|2016|         9|             2|            NULL|  Atur registrat|  1|la Nova Esquerra ...|     Eixample|  1255|Homes|  9|\n",
      "|2016|        10|             2|            NULL|  Atur registrat|  1|         Sant Antoni|     Eixample|   922|Homes| 10|\n",
      "+----+----------+--------------+----------------+----------------+---+--------------------+-------------+------+-----+---+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_new.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "9d3ff525-aeda-4e69-8473-8f34676d9df8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Any: string (nullable = true)\n",
      " |-- Codi_Barri: string (nullable = true)\n",
      " |-- Codi_Districte: string (nullable = true)\n",
      " |-- Demanda_ocupacio: string (nullable = true)\n",
      " |-- Demanda_ocupació: string (nullable = true)\n",
      " |-- Mes: string (nullable = true)\n",
      " |-- Nom_Barri: string (nullable = true)\n",
      " |-- Nom_Districte: string (nullable = true)\n",
      " |-- Nombre: string (nullable = true)\n",
      " |-- Sexe: string (nullable = true)\n",
      " |-- _id: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_new.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67533110-492c-4d8b-9a42-3ee845e54382",
   "metadata": {},
   "source": [
    "### 5. partition datas by any field for efficient query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "9182e9eb-bc02-40a3-9c2b-a487705796d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "repartitioned_df = df_new.repartition(\"Any\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4b8c2ec0-7f5e-4875-9443-36329271c213",
   "metadata": {},
   "source": [
    "### 6. save to parquet file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "682a1eec-e450-45d0-a49c-295ef9a0f515",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_path = \"./Formatted Zone/unemployment.parquet\"\n",
    "\n",
    "repartitioned_df.write.mode(\"overwrite\").parquet(output_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2adb3b40-a02b-4e78-92e7-354090f7a32e",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.19"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
